#include "app_pool.h"
#include <mqueue.h>
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <sys/types.h>
#include <log/log.h>

static char *MQ_NAME = "/app_pool_mq"; // 消息队列名称
static mqd_t mq;                       // 消息队列标识
static pthread_t *threads;             // 线程池中所有线程标识的容器
static int thread_num;                 // 线程数量

/**
 * @brief 线程函数，不断循环接收任务并执行,如果没有任务则阻塞等待
 */
static void *thread_func(void *arg)
{
    Task task;
    // 不断循环接收任务并执行,如果没有任务则阻塞等待
    while (1)
    {
        // 从消息队列中获取任务消息
        int len = mq_receive(mq, (char *)&task, sizeof(Task), NULL);
        // 执行任务
        if (len == sizeof(Task))
        {
            task.task_fun(task.arg);
        }
    }
}

int app_pool_init(int size)
{
    // 初始化任务(消息)队列
    struct mq_attr attr;
    attr.mq_maxmsg = 10;            // 消息队列中最大消息数
    attr.mq_msgsize = sizeof(Task); // 每个消息的最大字节数
    mq = mq_open(MQ_NAME, O_CREAT | O_RDWR, 0644, &attr);
    if (mq == -1)
    {
        perror("mq_open");
        return -1;
    }

    // 初始化线程池
    thread_num = size;
    threads = (pthread_t *)malloc(sizeof(pthread_t) * size);
    for (int i = 0; i < size; i++)
    {
        pthread_create(&threads[i], NULL, thread_func, NULL);
    }

    log_debug("线程池初始化成功");

    return 0;
}

int app_pool_destroy()
{
    // 关闭并删除消息队列
    mq_close(mq);
    mq_unlink(MQ_NAME);
    // 销毁线程池
    for (int i = 0; i < thread_num; i++)
    {
        pthread_cancel(threads[i]);
        pthread_join(threads[i], NULL);
    }

    free(threads);
}

int app_pool_registerTask(int (*task_fun)(void *arg), void *arg)
{
    // 创建任务
    Task task = {
        .task_fun = task_fun,
        .arg = arg
    };
    // 将任务发送到消息队列中
    return mq_send(mq, (char *)&task, sizeof(Task), 0);
}